package cn.jly.flink;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * 服务器上运行
 * bin/flink run -c cn.jly.flink.StreamWordCount -p 1 /opt/test/flink/flink-starter-1.0.0.jar --host node01 --port 9999
 * 命令行提交任务
 * <p>
 * bin/flink list 查看哪些正在运行
 * bin/flink list -a 查看所有，运行的和取消的等
 * <p>
 * bin/flink cancel jobId 取消正在执行的任务
 *
 * @PackageName cn.jly.flink
 * @ClassName StreamWordCount
 * @Description 流式word count案例
 * @Author 姬岚洋
 * @Date 2021/1/6 下午 10:27
 */
public class StreamWordCount {
    public static void main(String[] args) throws Exception {
        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
        final String host = parameterTool.get("host", "192.168.172.101");
        final String port = parameterTool.get("port", "9999");

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 监听端口 source
        final DataStreamSource<String> source = env.socketTextStream(host, Integer.parseInt(port), "\n");

        // transform
        final SingleOutputStreamOperator<WordAndCount> reduce = source
                .flatMap(new FlatMapFunction<String, WordAndCount>() {
                    @Override
                    public void flatMap(String s, Collector<WordAndCount> collector) throws Exception {
                        final String[] words = s.split("\\W+");
                        for (String word : words) {
                            if (StringUtils.isNotEmpty(word)) {
                                collector.collect(new WordAndCount(word, 1L));
                            }
                        }
                    }
                })
                .keyBy(new KeySelector<WordAndCount, String>() {
                    @Override
                    public String getKey(WordAndCount wordAndCount) throws Exception {
                        return wordAndCount.word;
                    }
                })
                .window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(4)))
                .reduce(new ReduceFunction<WordAndCount>() {
                    @Override
                    public WordAndCount reduce(WordAndCount wordAndCount, WordAndCount t1) throws Exception {
                        return new WordAndCount(wordAndCount.word, wordAndCount.count + t1.count);
                    }
                });

        // sink
        reduce.print().setParallelism(1);

        // execute
        env.execute("stream word count");
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class WordAndCount {
        private String word;
        private long count;
    }
}
